package main

import (
	"fmt"
	"strings"
	"time"

	"github.com/Shopify/sarama"
)

type KafkaHandler struct {
	topic    string
	producer sarama.AsyncProducer
	offset   int64
}

func NewKafkaHandler(urls, topic string) (KafkaHandler, error) {
	handler := KafkaHandler{}
	handler.topic = topic

	producerConfig := sarama.NewConfig()
	producerConfig.Producer.Return.Successes = true
	producerConfig.Producer.Timeout = 60 * time.Second

	producer, err := sarama.NewAsyncProducer(strings.Split(urls, ","), producerConfig)
	if err != nil {
		return handler, fmt.Errorf("NewKafkaHandler: Failed to create Kafka producer: %v", err)
	}
	handler.producer = producer
	return handler, nil
}

func (kh *KafkaHandler) Close() error {
	return kh.producer.Close()
}

func (kh *KafkaHandler) Produce(data string) (int64, error) {
	msg := &sarama.ProducerMessage{
		Timestamp: time.Now(),
		Topic:     kh.topic,
		Value:     sarama.StringEncoder(data),
	}
	go func() {
		kh.producer.Input() <- msg
	}()
	select {
	case err := <-kh.producer.Errors():
		if err != nil {
			return kh.offset, err
		}
	case msg := <-kh.producer.Successes():
		kh.offset = msg.Offset
	}
	return kh.offset, nil
}
